package com.asp.bigdata.flink.kafka2hdfs.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.asp.bigdata.flink.kafka2hdfs.pojo.Event;
import com.zyhlw.DataEncrypt.des.CipherPBEMd5Des;
import io.krakens.grok.api.Grok;
import io.krakens.grok.api.GrokCompiler;
import io.krakens.grok.api.Match;
import org.apache.commons.lang3.StringEscapeUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.io.Serializable;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class CustomUtils implements Serializable {
    private static final String PWD = "zxc@._875";

    public SingleOutputStreamOperator<Event> dealLogreport(DataStreamSource<String> streamSource, Properties jobConfig) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone(jobConfig.getProperty("time.zone", "GMT+08:00")));
        return streamSource.flatMap(
                (FlatMapFunction<String, Event>) (value, out) -> {
                    Map<String, String> valueMap = (Map) JSON.parse(value);
                    long time;
                    String message = "";
                    String regexStr = jobConfig.getProperty("split.regex");
                    String type = jobConfig.getProperty("topic");
                    message = valueMap.get("message");
                    try {
                        String[] strs = message.split(regexStr,13);
                        String even_time = strs[7];
                        time = simpleDateFormat.parse(even_time).getTime();

                    } catch (Exception e) {
                        e.printStackTrace();
                        time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                    }

                    // 获取type字段用于分区(表)，如果没有则属于脏数据，使用topic代替
                    type = valueMap.getOrDefault("type", type);
                    if (type.equals("logreport")) {
                        String[] fields = message.split("[|]", 13);
                        String datajson = fields[fields.length - 1];
                        JSONObject datajsonobject = JSON.parseObject(datajson);
                        String sdklog = datajsonobject.getString("sdklog");
                        if (sdklog.toString().startsWith("[")) {
                            JSONArray jsArr = JSON.parseArray(sdklog);
                            Iterator<Object> it = jsArr.iterator();
                            while (it.hasNext()) {
                                JSONObject ob = (JSONObject) it.next();
                                JSONObject sdkJson = new JSONObject();
                                if ((ob.toString()).contains("unKnownField")) {
                                    JSONObject sdkLogStr = JSON.parseObject(ob.getString("unKnownField"));
                                    sdkJson.put("sdklog", sdkLogStr);
                                } else {
                                    sdkJson.put("sdklog", ob);
                                }
                                datajsonobject.put("sdklog", sdkJson.getString("sdklog"));
                            }
                        }
                        datajson = datajsonobject.toJSONString();
                        StringBuffer sb = new StringBuffer();
                        for (int i = 0; i < fields.length - 1; i++) {
                            sb.append(fields[i]).append("|");
                        }
                        message = sb.append(datajson).toString();
                        out.collect(new Event(message, time, type));
                    }else if (type.equals("freeflowsdk")) {
                        //去掉转义符
                        message = StringEscapeUtils.unescapeJava(message);
                        //替换掉json部分的竖线
                        String[] fields = message.split("[|]", 13);
                        String jsonsdklog = fields[fields.length - 1];
                        jsonsdklog = jsonsdklog.replace('|', '&');
                        StringBuffer sb = new StringBuffer();
                        for (int i = 0; i < fields.length - 1; i++) {
                            sb.append(fields[i]).append("|");
                        }
                        message = sb.append(jsonsdklog).toString();
                        out.collect(new Event(message, time, type));
                    }else{
                        out.collect(new Event(message, time, type));
                    }
                });
    }
    public SingleOutputStreamOperator<Event> dealOpenapilog(DataStreamSource<String> streamSource, Properties jobConfig) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone(jobConfig.getProperty("time.zone", "GMT+08:00")));
        return streamSource.flatMap(
                (FlatMapFunction<String, Event>) (value, out) -> {
                    Map<String, String> valueMap = (Map) JSON.parse(value);
                    long time;
                    String message = "";
                    String regexStr = jobConfig.getProperty("split.regex");
                    String type = jobConfig.getProperty("topic");
                    message = valueMap.get("message");


                    // 获取type字段用于分区(表)，如果没有则属于脏数据，使用topic代替
                    type = valueMap.getOrDefault("type", type);

                    String notDecryptTypes = jobConfig.getProperty("notDecryptTypes");
                    if (notDecryptTypes != null && !notDecryptTypes.isEmpty()) {
                        for (String typeDe : notDecryptTypes.split(",")) {
                            if (type.equals(typeDe)) {
                                try {
                                    //String even_time_index = valueMap.get(jobConfig.getProperty("time.field.index"));
                                    //int index = Integer.valueOf(even_time_index);
                                    String[] strs = message.split(regexStr,13);
                                    String even_time = strs[0];
                                    time = simpleDateFormat.parse(even_time).getTime();

                                } catch (Exception e) {
                                    e.printStackTrace();
                                    time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                                }
                                if (type.equals("openapilog")) {
                                    String[] fields = message.split("[|]", 13);
                                    String jsonsdklog = fields[fields.length - 1];
                                    jsonsdklog = jsonsdklog.replace('|', ',');
                                    StringBuffer sb = new StringBuffer();
                                    for (int i = 0; i < fields.length - 1; i++) {
                                        sb.append(fields[i]).append("|");
                                    }
                                    message = sb.append(jsonsdklog).toString();
                                    out.collect(new Event(message, time, typeDe));
                                }else if (type.equals("rhapi-athena")){
                                    String[] fields = message.split("[|]", 8);
                                    String jsonsdklog = fields[fields.length - 1];
                                    jsonsdklog = jsonsdklog.replace('|', '&');
                                    StringBuffer sb = new StringBuffer();
                                    for (int i = 0; i < fields.length - 1; i++) {
                                        sb.append(fields[i]).append("|");
                                    }
                                    message = sb.append(jsonsdklog).toString();
                                    out.collect(new Event(message, time, typeDe));
                                }else{
                                    out.collect(new Event(message, time, typeDe));
                                }
                            }
                        }
                    }
                });
    }

    public SingleOutputStreamOperator<Event> dealUmchysdk(DataStreamSource<String> streamSource, Properties jobConfig) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone(jobConfig.getProperty("time.zone", "GMT+00:00")));
        return streamSource.flatMap(
                (FlatMapFunction<String, Event>) (value, out) -> {
                    Map<String, String> valueMap = (Map) JSON.parse(value);

                    long time;
                    String message = "", even_time = "";
                    String regexStr = jobConfig.getProperty("split.regex");
                    String type = jobConfig.getProperty("topic");

                    // 获取type字段用于分区(表)，如果没有则属于脏数据，使用topic代替
                    type = valueMap.getOrDefault("type", type);
                    // 多对一，type合并
                    String mergeType = mergeType(jobConfig, type);
                    if (mergeType != null) {
                        type = mergeType;
                    }

                    String desPwd = jobConfig.getProperty("des.pwd");
                    String needDecryptTypes = jobConfig.getProperty("needDecryptTypes");
                    message = valueMap.get("message");


                    //去掉转义符
                    //message = StringEscapeUtils.unescapeJava(message);
                    //替换掉json部分的竖线
                    String[] fields = message.split("[|]", 13);
                    String jsonsdklog = fields[fields.length - 1];
                    jsonsdklog = jsonsdklog.replace('|', '&');
                    StringBuffer sb = new StringBuffer();
                    for (int i = 0; i < fields.length - 1; i++) {
                        sb.append(fields[i]).append("|");
                    }
                    message = sb.append(jsonsdklog).toString();
                    //替换掉json数据中不规范的引号和转义符
                    message = message.replace("\\","").replace("\"{","{").replace("}\"","}");


                    String[] valueArr = message.split(regexStr);

                    // needn't decrypt types time
                    String notDecryptTypes = jobConfig.getProperty("notDecryptTypes");
                    if (notDecryptTypes != null && !notDecryptTypes.isEmpty()) {
                        for (String typeDe : notDecryptTypes.split(",")) {
                            if (type.equals(typeDe)) {

                                try {
                                    if (jobConfig.getProperty("time.field.index") != null) {
                                        even_time = valueArr[Integer.valueOf(jobConfig.getProperty("time.field.index"))];
                                    } else {
                                        even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                                    }
                                    time = simpleDateFormat.parse(even_time).getTime();

                                } catch (Exception e) {
                                    e.printStackTrace();
                                    time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                                }

                                out.collect(new Event(message, time, type));
                            }
                        }
                    }


                    // need decrypt types 有decryptType就指定输出，没有直接topic输出
                    if (needDecryptTypes != null && !needDecryptTypes.isEmpty() && desPwd != null) {
                        // 有decryptType就指定输出，没有直接topic输出
                        Map<String, String> fieldMap = (Map) JSON.parse(needDecryptTypes);
                        String needDecryptType, needDecryntIndex;

                        for (Map.Entry<String, String> str2Map : fieldMap.entrySet()) {
                            needDecryptType = str2Map.getKey();

                            if (needDecryptType.equals(type)) {
                                needDecryntIndex = str2Map.getValue();

                                for (String des_index : needDecryntIndex.split(",")) {
                                    String pre_data = valueArr[Integer.valueOf(des_index)];
                                    if (!pre_data.equals("") && !pre_data.equals("NULL") && !pre_data.equals("-") && !pre_data.equals("null")) {
                                        if (des_index != "0" || des_index != String.valueOf(valueArr.length)) {
                                            String realPwd = CipherPBEMd5Des.decrypt(desPwd, PWD);

                                            String des_data = CipherPBEMd5Des.encrypt(pre_data, realPwd);

                                            StringBuffer preBuffer = new StringBuffer(pre_data).append(regexStr).insert(0, regexStr);
                                            StringBuffer desBuffer = new StringBuffer(des_data).append(regexStr).insert(0, regexStr);

                                            message = message.replaceFirst(preBuffer.toString(), desBuffer.toString());
                                        } else {
                                            throw new Exception("fail to encrypting data！reason：加密字段在首尾");
                                        }
                                    }
                                }

                                try {
                                    if (jobConfig.getProperty("time.field.index") != null) {
                                        even_time = valueArr[Integer.valueOf(jobConfig.getProperty("time.field.index"))];
                                    } else {
                                        even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                                    }
                                    time = simpleDateFormat.parse(even_time).getTime();

                                } catch (Exception e) {
                                    e.printStackTrace();
                                    time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                                }

                                out.collect(new Event(message, time, type));
                                break;
                            }
                        }
                    } else if (notDecryptTypes == null && needDecryptTypes == null) {

                        try {
                            if (jobConfig.getProperty("time.field.index") != null) {
                                even_time = valueArr[Integer.valueOf(jobConfig.getProperty("time.field.index"))];
                            } else {
                                even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                            }
                            time = simpleDateFormat.parse(even_time).getTime();

                        } catch (Exception e) {
                            e.printStackTrace();
                            time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                        }

                        out.collect(new Event(message, time, type));
                    }

                });
    }

    public SingleOutputStreamOperator<Event> dealUmcrhsdk(DataStreamSource<String> streamSource, Properties jobConfig) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone(jobConfig.getProperty("time.zone", "GMT+00:00")));
        return streamSource.flatMap(
                (FlatMapFunction<String, Event>) (value, out) -> {
                    Map<String, String> valueMap = (Map) JSON.parse(value);

                    long time;
                    String message = "", even_time = "";
                    String regexStr = jobConfig.getProperty("split.regex");
                    String type = jobConfig.getProperty("topic");

                    // 获取type字段用于分区(表)，如果没有则属于脏数据，使用topic代替
                    type = valueMap.getOrDefault("type", type);
                    // 多对一，type合并
                    String mergeType = mergeType(jobConfig, type);
                    if (mergeType != null) {
                        type = mergeType;
                    }

                    String desPwd = jobConfig.getProperty("des.pwd");
                    String needDecryptTypes = jobConfig.getProperty("needDecryptTypes");
                    message = valueMap.get("message");

                    //去掉转义符
                    //message = StringEscapeUtils.unescapeJava(message);
                    //替换掉json部分的竖线
                    String[] fields = message.split("[|]", 13);
                    String jsonsdklog = fields[fields.length - 1];
                    jsonsdklog = jsonsdklog.replace('|', '&');
                    StringBuffer sb = new StringBuffer();
                    for (int i = 0; i < fields.length - 1; i++) {
                        sb.append(fields[i]).append("|");
                    }
                    message = sb.append(jsonsdklog).toString();
                    //替换掉json数据中不规范的引号和转义符
                    message = message.replace("\\","").replace("\"{","{").replace("}\"","}");

                    String[] valueArr = message.split(regexStr);

                    // needn't decrypt types time
                    String notDecryptTypes = jobConfig.getProperty("notDecryptTypes");
                    if (notDecryptTypes != null && !notDecryptTypes.isEmpty()) {
                        for (String typeDe : notDecryptTypes.split(",")) {
                            if (type.equals(typeDe)) {

                                try {
                                    if (jobConfig.getProperty("time.field.index") != null) {
                                        even_time = valueArr[Integer.valueOf(jobConfig.getProperty("time.field.index"))];
                                    } else {
                                        even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                                    }
                                    time = simpleDateFormat.parse(even_time).getTime();

                                } catch (Exception e) {
                                    e.printStackTrace();
                                    time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                                }

                                out.collect(new Event(message, time, type));
                            }
                        }
                    }


                    // need decrypt types 有decryptType就指定输出，没有直接topic输出
                    if (needDecryptTypes != null && !needDecryptTypes.isEmpty() && desPwd != null) {
                        // 有decryptType就指定输出，没有直接topic输出
                        Map<String, String> fieldMap = (Map) JSON.parse(needDecryptTypes);
                        String needDecryptType, needDecryntIndex;

                        for (Map.Entry<String, String> str2Map : fieldMap.entrySet()) {
                            needDecryptType = str2Map.getKey();

                            if (needDecryptType.equals(type)) {
                                needDecryntIndex = str2Map.getValue();

                                for (String des_index : needDecryntIndex.split(",")) {
                                    String pre_data = valueArr[Integer.valueOf(des_index)];
                                    if (!pre_data.equals("") && !pre_data.equals("NULL") && !pre_data.equals("-") && !pre_data.equals("null")) {
                                        if (des_index != "0" || des_index != String.valueOf(valueArr.length)) {
                                            String realPwd = CipherPBEMd5Des.decrypt(desPwd, PWD);

                                            String des_data = CipherPBEMd5Des.encrypt(pre_data, realPwd);

                                            StringBuffer preBuffer = new StringBuffer(pre_data).append(regexStr).insert(0, regexStr);
                                            StringBuffer desBuffer = new StringBuffer(des_data).append(regexStr).insert(0, regexStr);

                                            message = message.replaceFirst(preBuffer.toString(), desBuffer.toString());
                                        } else {
                                            throw new Exception("fail to encrypting data！reason：加密字段在首尾");
                                        }
                                    }
                                }

                                try {
                                    if (jobConfig.getProperty("time.field.index") != null) {
                                        even_time = valueArr[Integer.valueOf(jobConfig.getProperty("time.field.index"))];
                                    } else {
                                        even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                                    }
                                    time = simpleDateFormat.parse(even_time).getTime();

                                } catch (Exception e) {
                                    e.printStackTrace();
                                    time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                                }

                                out.collect(new Event(message, time, type));
                                break;
                            }
                        }
                    } else if (notDecryptTypes == null && needDecryptTypes == null) {

                        try {
                            if (jobConfig.getProperty("time.field.index") != null) {
                                even_time = valueArr[Integer.valueOf(jobConfig.getProperty("time.field.index"))];
                            } else {
                                even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                            }
                            time = simpleDateFormat.parse(even_time).getTime();

                        } catch (Exception e) {
                            e.printStackTrace();
                            time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                        }

                        out.collect(new Event(message, time, type));
                    }

                });
    }

    public SingleOutputStreamOperator<Event> dealRcsMsg(DataStreamSource<String> streamSource, Properties jobConfig) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone(jobConfig.getProperty("time.zone","GMT+00:00")));
        return streamSource.flatMap((FlatMapFunction<String, Event>) (value, out) -> {
            Map<String, String> valueMap = (Map) JSON.parse(value);
            long time;
            String message = "";

            //String regexStr = jobConfig.getProperty("split.regex");
            String type = jobConfig.getProperty("topic");

            // 获取type字段用于分区(表)，如果没有则属于脏数据，使用topic代替
            type = valueMap.getOrDefault("type", type);
            if (valueMap.containsKey("message") && valueMap.get("message") != null) {
                message = valueMap.get("message").replace("\n","");
            }

            Date date = null;
            GrokCompiler grokCompiler = GrokCompiler.newInstance();
            grokCompiler.registerDefaultPatterns();

            //按模块解析数据
            if (type.equals("shortMessageProgram")) {
                if (message.contains("logType:")) {
                    //grokCompiler.registerDefaultPatterns();
                    grokCompiler.register("DATETIME", "\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}.\\d{3}");
                    Grok grok = grokCompiler.compile("^%{DATETIME:log_time} .*- (.*\\] |)%{DATA:msgBody}$");
                    Match grokMatch = grok.match(message);
                    Map<String, Object> resultMap = grokMatch.capture();
                    if (resultMap != null && !resultMap.isEmpty()) {
                        for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                            valueMap.put(entry.getKey(), entry.getValue().toString());
                        }
                        String[] field_list = valueMap.get("msgBody").split(",");
                        if (field_list != null && field_list.length != 0) {
                            for (int i = 0; i < field_list.length; i++) {
                                String str = field_list[i];
                                String[] strs = str.split(":", 2);
                                if (strs.length >= 2) {
                                    valueMap.put(strs[0], strs[1]);
                                } else {
                                    valueMap.put(strs[0], "");
                                }
                            }

                            if (valueMap != null && !valueMap.isEmpty()) {
                                if (valueMap.get("type").equals("callInterface")) {
                                    if (valueMap.containsKey("duration") && valueMap.get("duration") != null && !valueMap.get("duration").isEmpty()) {
                                        Matcher m = Pattern.compile("^[0-9]+").matcher(valueMap.get("duration"));
                                        if (m.find()) {
                                            valueMap.put("durations", m.group());
                                            valueMap.remove("duration");
                                        }
                                    }
                                }
                                if (valueMap.containsKey("time")) {
                                    valueMap.put("times", valueMap.remove("time"));
                                }
                            }
                            valueMap.replace("type", valueMap.get("logType"));
                            valueMap.remove("msgBody");
                        }
                    }

                    //将log_time解析为时间格式
                    if (valueMap.containsKey("log_time")) {
                        if (!valueMap.get("log_time").isEmpty() && valueMap.get("log_time") != null) {
                            try {
                                date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").parse(valueMap.get("log_time"));
                            } catch (Exception e) {
                                System.out.println(e.getMessage());
                            }
                        }
                    }
                }
            }else if (type.equals("a2p_hfx_nginx")) {
                //GrokCompiler grokCompiler = GrokCompiler.newInstance();
                // 进行注册, registerDefaultPatterns()方法注册的是Grok内置的patterns
                //grokCompiler.registerDefaultPatterns();
                // 添加自定的正则表达式，解析日期（例如：2021-07-07 13:45:39.210）
                //grokCompiler.register("DATETIME","[0-9,\\.\\-: ]+");

                Grok grok = grokCompiler.compile("^%{IP:remote_addr} - %{DATA:remote_user} \\[%{DATA:time_local} \\+0800\\] \\\"%{DATA:request}\\\" %{DATA:status} %{DATA:body_bytes_sent} \\\"%{DATA:http_referer}\\\" \\\"%{DATA:http_user_agent}\\\" \\\"%{DATA:http_x_forwarded_for}\\\" %{DATA:request_time} %{DATA:upstream_response_time}$");
                // 通过match()方法进行匹配, 对log进行解析, 按照指定的格式进行输出
                Match grokMatch = grok.match(message);
                Map<String, Object> resultMap = grokMatch.capture();
                if (resultMap != null && !resultMap.isEmpty()) {
                    for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                        //System.out.println( entry.getKey() +":"+ entry.getValue());
                        valueMap.put(entry.getKey(), entry.getValue().toString());
                    }
                }
                //将log_time解析为时间格式
                if (valueMap.containsKey("time_local")) {
                    if (!valueMap.get("time_local").isEmpty() && valueMap.get("time_local") != null) {
                        try {
                            date = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss", Locale.US).parse(valueMap.get("time_local"));
                        } catch (Exception e) {
                            System.out.println(e.getMessage());
                        }
                    }
                }
            }else if (type.equals("a2p_10285_nginx")) {
                //grokCompiler.registerDefaultPatterns();
                Grok grok = grokCompiler.compile("^%{DATA:http_x_forwarded_for}\\|%{DATA:time_iso8601}\\+08:00\\|%{DATA:request}\\|%{DATA:status}\\|%{DATA:body_bytes_sent}\\|%{DATA:http_referer}\\|%{DATA:http_user_agent}\\|%{DATA:remote_addr}\\|%{DATA:host}\\|%{DATA:hostname}\\|%{DATA:upstream_addr}\\|%{DATA:request_time}\\|%{DATA:upstream_response_time}$");
                // 通过match()方法进行匹配, 对log进行解析, 按照指定的格式进行输出
                Match grokMatch = grok.match(message);
                Map<String, Object> resultMap = grokMatch.capture();
                if (resultMap != null && !resultMap.isEmpty()) {
                    for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                        valueMap.put(entry.getKey(), entry.getValue().toString());
                    }
                }
                //将log_time解析为时间格式
                if (valueMap.containsKey("time_iso8601")) {
                    if (!valueMap.get("time_iso8601").isEmpty() && valueMap.get("time_iso8601") != null) {
                        try {
                            date = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss").parse(valueMap.get("time_iso8601"));
                        } catch (Exception e) {
                            System.out.println(e.getMessage());
                        }
                    }
                }
            }else if (type.equals("voice-notify")) {
                if (message.contains("logType:")) {
                    String msgBody = message.split(".*: ")[1];
                    if (msgBody != null && msgBody.isEmpty()) {
                        String[] field_list = msgBody.split("\\|");
                        if (field_list != null && field_list.length != 0) {
                            for (int i = 0; i < field_list.length; i++) {
                                String str = field_list[i];
                                String[] strs = str.split(":", 2);
                                if (strs.length >= 2) {
                                    valueMap.put(strs[0], strs[1]);
                                } else {
                                    valueMap.put(strs[0], "");
                                }
                            }
                            //valueMap.put("local_time", valueMap.get("timestamp"));
                            valueMap.put("times", valueMap.remove("time"));
                            if (valueMap != null && !valueMap.isEmpty()) {
                                if (valueMap.get("logType").equals("getToken")) {
                                    valueMap.replace("type", "voice_" + valueMap.get("logType"));
                                } else {
                                    valueMap.replace("type", valueMap.get("logType"));
                                }
                            }
                        }
                    }
                    if (valueMap.containsKey("timestamp")) {
                        valueMap.put("local_time", valueMap.get("timestamp"));
                        if (!valueMap.get("local_time").isEmpty() && valueMap.get("local_time") != null) {
                            try {
                                date = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'").parse(valueMap.get("local_time"));
                            } catch (Exception e) {
                                System.out.println(e.getMessage());
                            }
                        }
                    }
                }
            }else if (type.equals("vms-server")) {
                if (message.contains("logType:")) {
                    //grokCompiler.registerDefaultPatterns();
                    grokCompiler.register("DATETIME", "\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}.\\d{3}");
                    Grok grok = grokCompiler.compile("^%{DATETIME:local_time} .*: %{DATA:msgBody}$");
                    Match grokMatch = grok.match(message);
                    Map<String, Object> resultMap = grokMatch.capture();
                    if (resultMap != null && !resultMap.isEmpty()) {
                        for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                            valueMap.put(entry.getKey(), entry.getValue().toString());
                        }
                        String[] field_list = valueMap.get("msgBody").split("\\|");
                        if (field_list != null && field_list.length != 0) {
                            for (int i = 0; i < field_list.length; i++) {
                                String str = field_list[i];
                                String[] strs = str.split(":", 2);
                                if (strs.length >= 2) {
                                    valueMap.put(strs[0], strs[1]);
                                } else {
                                    valueMap.put(strs[0], "");
                                }
                            }
                            valueMap.put("times", valueMap.remove("time"));
                            valueMap.put("actionType", valueMap.remove("type"));
                            valueMap.put("type", valueMap.get("logType"));
                            valueMap.remove("msgBody");
                        }
                    }
                    //将log_time解析为时间格式
                    if (valueMap.containsKey("local_time")) {
                        if (!valueMap.get("local_time").isEmpty() && valueMap.get("local_time") != null) {
                            try {
                                date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").parse(valueMap.get("local_time"));
                            } catch (Exception e) {
                                System.out.println(e.getMessage());
                            }
                        }
                    }
                }
            } else if (type.equals("voice-api") ) {
                if (message.contains("logType:")) {
                    //grokCompiler.registerDefaultPatterns();
                    grokCompiler.register("DATETIME", "\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}.\\d{3}");
                    Grok grok = grokCompiler.compile("^%{DATETIME:local_time} .*- %{DATA:msgBody}$");
                    Match grokMatch = grok.match(message);
                    Map<String, Object> resultMap = grokMatch.capture();
                    if (resultMap != null && !resultMap.isEmpty()) {
                        for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                            valueMap.put(entry.getKey(), entry.getValue().toString());
                        }
                        String[] field_list = valueMap.get("msgBody").split("\\|");
                        if (field_list != null && field_list.length != 0) {
                            for (int i = 0; i < field_list.length; i++) {
                                String str = field_list[i];
                                String[] strs = str.split(":", 2);
                                if (strs.length >= 2) {
                                    valueMap.put(strs[0], strs[1]);
                                } else {
                                    valueMap.put(strs[0], "");
                                }
                            }
                            valueMap.put("times", valueMap.remove("time"));
                            valueMap.put("actionType", valueMap.remove("type"));
                            valueMap.put("type", valueMap.get("logType"));
                            valueMap.remove("msgBody");
                        }
                    }
                    if (valueMap.containsKey("local_time")) {
                        if (!valueMap.get("local_time").isEmpty() && valueMap.get("local_time") != null) {
                            try {
                                date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").parse(valueMap.get("local_time"));
                            } catch (Exception e) {
                                System.out.println(e.getMessage());
                            }
                        }
                    }
                }
            }

            //将log_time解析为yyyy-MM-dd'T'HH:mm:ss.SSS'Z'的时间格式
            //SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
            if (date != null) {
                valueMap.put("@timestamp", simpleDateFormat.format(date));
            }
            try {
                String even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                time = simpleDateFormat.parse(even_time).getTime();
                //time = format.parse(valueMap.get("@timestamp")).getTime();
            } catch (Exception e) {
                e.printStackTrace();
                time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
            }

            //无需加密的模块直接输出
            String notDecryptTypes = jobConfig.getProperty("notDecryptTypes");
            if (notDecryptTypes != null && !notDecryptTypes.isEmpty()) {
                for (String typeDe : notDecryptTypes.split(",")) {
                    if (typeDe.equals(valueMap.get("type"))){
                        value = JSON.toJSONString(valueMap);
                        out.collect(new Event(value, time, typeDe));
                        break;
                    }
                }
            }

            //需要加密的模块，使用CipherPBEMd5Des加密
            String desPwd = jobConfig.getProperty("des.pwd");
            String needDecryptTypes = jobConfig.getProperty("needDecryptTypes");

            if (needDecryptTypes != null && !needDecryptTypes.isEmpty() && desPwd != null) {
                Map<String,String> fieldMap = (Map)JSON.parse(needDecryptTypes);
                String needDecryptType, needDecryntFields;

                for (Map.Entry<String, String> str2Map : fieldMap.entrySet()) {

                    needDecryptType = str2Map.getKey();

                    if (needDecryptType.equals(valueMap.get("type"))) {
                        needDecryntFields = str2Map.getValue();
                        for (String desField : needDecryntFields.split(",")) {
                            if (valueMap.get(desField) != null) {
                                String realPwd = CipherPBEMd5Des.decrypt(desPwd,PWD);
                                String encrypt_data = CipherPBEMd5Des.encrypt(valueMap.get(desField), realPwd);
                                valueMap.replace(desField, encrypt_data);
                            }
                        }
                        value = JSON.toJSONString(valueMap);
                        out.collect(new Event(value, time, needDecryptType));
                        break;
                    }
                }
            }else if(notDecryptTypes == null && needDecryptTypes == null){
                value = JSON.toJSONString(valueMap);
                out.collect(new Event(value, time, valueMap.get("type")));
            }
        });
    }

    private String mergeType(Properties jobConfig, String type) {
        String mergeType = jobConfig.getProperty("mergeType");
        if (mergeType != null) {
            String[] split = mergeType.split("\\|");
            for (String typeMer : split[0].split(",")) {
                if (type.equals(typeMer)) {
                    return split[1];
                }
            }
        }
        return null;
    }
}
